package com.lagou.util;

import com.lagou.client.RPCConsumer;
import com.lagou.client.ServerInfo;
import com.lagou.handler.UserClientHandler;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.io.IOException;
import java.io.InputStream;
import java.util.*;
import java.util.stream.Collectors;

/**
 * zk工具类：1,获取可用的服务端
 */
public class ZkUtils {

    private static Map<String, UserClientHandler> userClientHandlerMap = new HashMap<>();


    /**
     * 加载zk服务节点信息并存储
     */
    public static void loadZkProvider() throws Exception {
        //加载配置文件
        InputStream inputStream = ZkUtils.class.getClassLoader().getResourceAsStream("application.properties");
        Properties properties=new Properties();
        properties.load(inputStream);
        String zkServerUrl = properties.getProperty("zk.zkServerUrl");
        String zkServerRootNode = properties.getProperty("zk.zkRootPath");
        System.out.println("服务url："+zkServerUrl);
        System.out.println("节点根路径："+zkServerRootNode);

        //连接zk服务器
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 5);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(zkServerUrl)
                .sessionTimeoutMs(1000)
                .connectionTimeoutMs(5000)
                .retryPolicy(retryPolicy)
                .build();
        client.start();

        //获取server节点
        List<String> serverNodeList = client.getChildren().forPath(zkServerRootNode);
        for (String serverNodePath : serverNodeList) {
            byte[] bytes = client.getData().forPath(zkServerRootNode + "/" + serverNodePath);
            String[] serverNodeInfo = new String(bytes).split(":");
            buildClientHandler(serverNodeInfo[0], Integer.parseInt(serverNodeInfo[1]));
        }

        //添加server节点变更器，监听server服务器的加入和下线
        PathChildrenCache cache = new PathChildrenCache(client, zkServerRootNode, true);
        cache.start();
        PathChildrenCacheListener serverNodeListener = (server, event) -> {
            //检测到新节点加入，把新的server加入到处理器集合中
            String[] serverNodeInfo = new String(event.getData().getData()).split(":");
            if(serverNodeInfo.length>0){
                ServerInfo serverInfo=new ServerInfo(serverNodeInfo[0],Integer.parseInt(serverNodeInfo[1]));
                if (PathChildrenCacheEvent.Type.CHILD_ADDED.equals(event.getType())) {
                    System.out.println("有新的server节点加入，ip=" + serverNodeInfo[0] + ",port=" + serverNodeInfo[1]);
                    buildClientHandler(serverNodeInfo[0], Integer.parseInt(serverNodeInfo[1]));

                }
                if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(event.getType())) {
                    //检测到节点删除，把server从处理器集合中移除
                    //String[] serverNodeInfo = new String(event.getData().getData()).split(":");
                    System.out.println("server节点下线，ip=" + serverNodeInfo[0] + ",port=" + serverNodeInfo[1]);
                    userClientHandlerMap.remove(serverNodeInfo[0] + serverNodeInfo[1]);
                }

            }

        };
        cache.getListenable().addListener(serverNodeListener);
    }

    /**
     * @param serverIp
     * @param port
     * @throws InterruptedException
     */
    private static void buildClientHandler(String serverIp, int port) throws InterruptedException {
        ServerInfo serverInfo = new ServerInfo(serverIp, port);

        UserClientHandler userClientHandler = RPCConsumer.initClient(serverInfo);
        userClientHandlerMap.put(serverIp + port, userClientHandler);
    }



    /**
     * 获取可用的服务端
     * @return
     */
    public static UserClientHandler getAvalibleServer(){
        if (userClientHandlerMap.isEmpty()) {
            return null;
        }

        //用来存放可用的服务列表
        List<ServerInfo> serverList = new ArrayList<>();


        //从服务列表中  选取响应时间最短的server
        for (String serverIpPort : userClientHandlerMap.keySet()) {
            ServerInfo serverInfo = userClientHandlerMap.get(serverIpPort).getServerInfo();
            long currentTimeMillis = System.currentTimeMillis();
            //判断最后一次响应时间是否超过5秒，超过5秒则清零
            if (currentTimeMillis - serverInfo.getLastCallTime() >= 5000) {
                serverInfo.setSpendTime(0L);
            }else{
                serverInfo.setSpendTime(currentTimeMillis - serverInfo.getLastCallTime());
            }
            serverList.add(serverInfo);
        }

        //按响应时间排序，选取响应时间最短的server
        List<ServerInfo> sortedServerList = serverList
                .stream().sorted(Comparator.comparing(ServerInfo::getSpendTime)).collect(Collectors.toList());
        ServerInfo fastServerInfo = sortedServerList.get(0);


        return userClientHandlerMap.get(fastServerInfo.getIp()+fastServerInfo.getPort());
    }
}
